package FlumeConfi;

import org.apache.commons.io.FileUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.EventDrivenSource;
import org.apache.flume.channel.ChannelProcessor;
import org.apache.flume.conf.Configurable;
import org.apache.flume.event.EventBuilder;
import org.apache.flume.source.AbstractSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;

import java.nio.charset.Charset;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;


public class CustomSource extends AbstractSource implements Configurable ,EventDrivenSource {

    //记录日志
    private String logFile;
    private String offsetFile;
    private String  charset;
    private String interval;
    private ExecutorService pool;
    private FileRunnable fileRunnable;
    private static final Logger logger = LoggerFactory.getLogger(CustomSource.class);

    //读取配置文件（flume在执行一次job时定义的配置文件）
    @Override
    public void configure(Context context) {

        //获取配置的参数
        //根据Context读取配置，Context会自动加载flume启动时指定的配置
        //这是真正日志文件
        logFile = context.getString("logFile");
        //获取偏移量文件
        offsetFile = context.getString("offsetFile");
        //获取轮询间隔,读配置文件间隔时间，默认值1000ms
        interval = context.getString("interval");
        //默认使用utf-8
        charset = context.getString("charset","UTF-8");
    }

    //1、创建一个线程来监听一个文件
    @Override
    public synchronized void start() {
        //创建一个单线程的线程池
        pool = Executors.newSingleThreadExecutor();
        //获取一个ChannelProcessor
        ChannelProcessor channelProcessor = getChannelProcessor();

        fileRunnable = new FileRunnable(logFile,offsetFile,interval,charset,channelProcessor);
        //提交到线程池中
        pool.submit(fileRunnable);

        super.start();
    }

    @Override
    public synchronized void stop() {
        //停止线程池
        fileRunnable.setFlag(false);
        pool.shutdown();


        super.stop();
    }

    private static class FileRunnable implements Runnable{

        private String logFile;
        private String  offsetFile;
        private String interval;
        private String charset;

        private ChannelProcessor channelProcessor;
        private long offset = 0L;
        private RandomAccessFile raf;
        private File file;
        private Boolean flag = true;

        public void setFlag(Boolean flag) {
            this.flag = flag;
        }

        //将外部类的参数传进内部静态类
        public FileRunnable(String logFile, String offsetFile, String interval, String charset, ChannelProcessor channelProcessor) {
            this.logFile = logFile;
            this.offsetFile = offsetFile;
            this.interval = interval;
            this.charset = charset;
            this.channelProcessor = channelProcessor;
            //判断咱们的偏移量文件存在不存在
            file = new File(offsetFile);
            if(!file.exists()){
                try {
                    file.createNewFile();
                } catch (IOException e) {
                    logger.error("createNewFile failed",e);
                }
            }
            //读取文件,读出了一个偏移量
            try {
                String offsetStr = FileUtils.readFileToString(file);
                //将字符串转成long类型
                offset = Long.parseLong(offsetStr);
            } catch (Exception e) {
                logger.error("parseLong failed");
            }

            //seek方法，就是将咱们的位置直接到达偏移量指定的位置
            //RandomAccessFile
            try {
                raf = new RandomAccessFile(logFile, "r");
                //将raf挪个位置
                raf.seek(offset);
            } catch (Exception e) {
                logger.error("seek failed",e);
            }

        }
        @Override
        public void run() {
            while (flag){
                try {
                    String line =raf.readLine();
                    if(StringUtils.isNotEmpty(line)){
                        //将数据封装成event类型
                        Event event = EventBuilder.withBody(line, Charset.forName(charset));
                        //发送给channel
                        channelProcessor.processEvent(event);
                        //获取偏移量,更新偏移量
                        offset = raf.getFilePointer();
                        //将偏移量写入到位置文件中
                        FileUtils.writeStringToFile(file,String.valueOf(offset));
                    }else{
                        //没读到睡一会儿
                        Thread.sleep(Long.parseLong(interval));
                    }

                } catch (IOException e) {
                    logger.error("IO failed",e);
                } catch (InterruptedException e) {
                    logger.error("Interrupted failed",e);
                }
            }
        }
    }
}
